package com.chinamobile.iot.command;

import com.chinamobile.iot.config.MqttConfiguration;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by szl on 2017/5/27.
 */
@Component
public class CommandCache {

    private static Logger logger = LoggerFactory.getLogger(CommandCache.class);

    /**
     * CommandKey ---> CommandToken
     * 用于存储已经下发给设备的命令信息
     * 默认命令信息缓存60秒，默认有效期为60秒
     */
    private Cache<CommandKey, String> cacheCommand;
    /**
     * CommandKey --> CommandValue
     * 用于存储已经收到的响应信息，默认有效期为3秒
     */
    private Cache<CommandKey, String> cacheAck;

    @Autowired
    private MqttConfiguration mqttConfiguration;

    @Autowired
    private CommandCacheTimeoutLister commandCacheTimeoutLister;
    @Autowired
    private CommandAckCacheTimeoutListener commandAckCacheTimeoutListener;

    private ScheduledExecutorService service;

    @PostConstruct
    public void init() {
        //初始化命令缓存
        cacheCommand = Caffeine.newBuilder()
                .expireAfterWrite(mqttConfiguration.getCmdTimeout(), TimeUnit.SECONDS)
                .removalListener(commandCacheTimeoutLister)
                .build();

        // cacheAck超时时,会操作cacheCommand, 所以这里进行设置
        commandAckCacheTimeoutListener.setCacheCommand(cacheCommand);

        // 初始化CacheAck
        cacheAck = Caffeine.newBuilder()
                .expireAfterWrite(2, TimeUnit.SECONDS)
                .removalListener(commandAckCacheTimeoutListener)
                .build();

        // 每秒启动一次后台超时检查任务
        service = Executors.newScheduledThreadPool(1);
        service.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                cacheCommand.cleanUp();
                cacheAck.cleanUp();
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    @PreDestroy
    public void destroy() {
        logger.info("CommandCache destroy is called.");
        service.shutdown();
    }

    public void addCommand(String clientId, String deviceId, Integer msgId, String commandToken) {
        CommandKey key = new CommandKey();
        key.setMsgId(msgId);
        key.setMqttId(deviceId);
        key.setClientId(clientId);
        cacheCommand.put(key, commandToken);
    }

    public void removeCommand(String clientId, String deviceId, Integer msgId) {
        CommandKey key = new CommandKey();
        key.setMsgId(msgId);
        key.setMqttId(deviceId);
        key.setClientId(clientId);
        cacheCommand.invalidate(key);
    }

    public String getCommandId(String clientId, String deviceId, Integer msgId) {
        CommandKey key = new CommandKey();
        key.setMsgId(msgId);
        key.setMqttId(deviceId);
        key.setClientId(clientId);

        String value = cacheCommand.getIfPresent(key);
        return value;
    }

    public void addAck(String clientId, String deviceId, Integer msgId, String commandToken) {
        CommandKey key = new CommandKey();
        key.setMsgId(msgId);
        key.setMqttId(deviceId);
        key.setClientId(clientId);
        cacheAck.put(key, commandToken);
    }

    public void removeAck(String clientId, String deviceId, Integer msgId) {
        CommandKey key = new CommandKey();
        key.setMsgId(msgId);
        key.setMqttId(deviceId);
        key.setClientId(clientId);
        cacheAck.invalidate(key);
    }

}
